package org.lionsoul.websnail.worker;

import org.lionsoul.websnail.task.Task;

import java.util.concurrent.*;


/**
 * Spider worker
 * 实现页面的下载和分析功能
 * LocalWorker and RemoteWorker could base on this
 *
 * @author chenxin<chenxin619315 @ gmail.com>
 */
public class SpiderWorker implements Worker {
    /**
     * current running state of the worker
     */
    private volatile int state = Worker.READY;
    private final Object LOCK = new Object();

    // 当前 Worker 没有线程可用时，sleep 的时间(毫秒)
    private final int busySleepTime;

    /**
     * thread pool
     */
    private final ThreadPoolExecutor threadPool;

    /**
     * task queue
     */
    private final BlockingQueue<Task> taskQueue;

    /**
     * consumer thread
     */
    private Thread consumerThread = null;

    /**
     * create and return the default worker
     *
     * @return SpiderWorker
     */
    public static SpiderWorker createDefault() {
        return new SpiderWorker(
                1500,
                50, // 核心线程数
                100, // 最大线程数
                6,
                new ArrayBlockingQueue<>(5000),
                new LinkedBlockingQueue<>(10000)
        );
    }

    /**
     * create a worker builder so you can define the worker
     *
     * @return SpiderWorkerBuilder
     */
    public static SpiderWorkerBuilder custom() {
        return new SpiderWorkerBuilder();
    }

    SpiderWorker(
            int busySleepTime,
            int coreThreadSize,
            int maxThreadSize,
            int threadAliveSec,
            BlockingQueue<Runnable> threadPoolQueue,
            BlockingQueue<Task> taskQueue) {
        this.busySleepTime = busySleepTime;
        this.taskQueue = taskQueue;

        //create the thread pool
        threadPool = new ThreadPoolExecutor(
                coreThreadSize, maxThreadSize, threadAliveSec,
                TimeUnit.SECONDS,
                threadPoolQueue,
                new ThreadPoolExecutor.DiscardOldestPolicy()
        );
    }

    /**
     * @see Worker#getLeftTask()
     */
    @Override
    public int getLeftTask() {
        return taskQueue.size();
    }

    /**
     * @see Worker#addTask(Task)
     */
    @Override
    public boolean addTask(Task task) {
        return taskQueue.offer(task);
    }

    /**
     * @see Worker#getAvailableThread()
     */
    @Override
    public int getAvailableThread() {
        return threadPool.getMaximumPoolSize() - threadPool.getActiveCount();
    }

    /**
     * @see Worker#getWorkingThread()
     */
    @Override
    public int getWorkingThread() {
        return threadPool.getActiveCount();
    }

    // 判断当前 Worker 是否还能处理更多的任务
    @Override
    public boolean couldHoldMore() {
        int leftThread = threadPool.getMaximumPoolSize() - threadPool.getActiveCount();
        return (leftThread > 0);
    }

    /**
     * @see Worker#process(Runnable)
     */
    @Override
    public boolean process(Runnable task) {
        if (couldHoldMore()) {
            threadPool.execute(task);
            return true;
        }

        return false;
    }

    // 启动爬虫 worker
    @Override
    public void start() {
        consumerThread = new Thread(() -> {
            while (state != Worker.STOPPED) {
                //check and handler the stopped operation

                //check and handler the pause operation
                if (state == Worker.PAUSED) {
                    synchronized (LOCK) {
                        try {
                            LOCK.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }

                try {
                    //check if the worker could hold more task
                    if (!couldHoldMore()) {
                        Thread.sleep(busySleepTime);
                        continue;
                    }

                    //take an url from the task queue and process it
                    Task task = taskQueue.take();
                    process(task);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        //start the task consumer thread
        consumerThread.start();
    }

    @Override
    public int getState() {
        return state;
    }

    @Override
    public void pause() {
        if (state == Worker.PAUSED) return;
        state = Worker.PAUSED;
    }

    @Override
    public void resume() {
        if (state == Worker.RUNNING) return;
        synchronized (LOCK) {
            state = Worker.RUNNING;
            LOCK.notify();
        }
    }

    /**
     * @see Worker#shutdown()
     */
    @Override
    public void shutdown() {
        state = Worker.STOPPED;
        taskQueue.clear();
        threadPool.shutdown();

        //interrupt the consumer thread
        if (consumerThread != null) {
            consumerThread.interrupt();
        }
    }
}
